<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="qPHzi" id="qPHzi"><span data-lake-id="ud7d35a5c" id="ud7d35a5c">典型回答</span></h1>
  <p data-lake-id="u6c5ccc84" id="u6c5ccc84"><br></p>
  <p data-lake-id="uf213ca0c" id="uf213ca0c"><span data-lake-id="u400f0caa" id="u400f0caa">高水位（HW，High Watermark）是Kafka中的一个重要的概念，主要是用于管理消费者的进度和保证数据的可靠性的。</span></p>
  <p data-lake-id="u4988628e" id="u4988628e"><span data-lake-id="ue0d2b4a8" id="ue0d2b4a8">​</span><br></p>
  <p data-lake-id="u30f88ef2" id="u30f88ef2"><span data-lake-id="u766a4a29" id="u766a4a29">高水位标识了一个特定的消息偏移量（offset），即一个分区中已提交消息的最高偏移量（offset），消费者只能拉取到这个 offset 之前的消息。消费者可以通过跟踪高水位来确定自己消费的位置。</span></p>
  <p data-lake-id="u4971a7a9" id="u4971a7a9"><span data-lake-id="ufc3af7f3" id="ufc3af7f3">​</span><br></p>
  <blockquote data-lake-id="uc3b298ff" id="uc3b298ff">
   <p data-lake-id="ucccd2216" id="ucccd2216"><span data-lake-id="u4991d760" id="u4991d760">这里的已提交指的是ISRs中的所有副本都记录了这条消息</span></p>
  </blockquote>
  <p data-lake-id="u8ddd7b30" id="u8ddd7b30"><span data-lake-id="u288108b9" id="u288108b9">​</span><br></p>
  <p data-lake-id="u71e670c0" id="u71e670c0"><span data-lake-id="ufd9e6d84" id="ufd9e6d84">在Kafka中，HW主要有两个作用：</span></p>
  <p data-lake-id="u3651a280" id="u3651a280"><span data-lake-id="u88aee1cf" id="u88aee1cf">​</span><br></p>
  <ul list="u22ddadeb">
   <li fid="ub938c195" data-lake-id="uede9e828" id="uede9e828"><span data-lake-id="uef83c4ca" id="uef83c4ca">消费进度管理：消费者可以通过记录上一次消费的偏移量，然后将其与分区的高水位进行比较，来确定自己的消费进度。消费者可以在和高水位对比之后继续消费新的消息，确保不会错过任何已提交的消息。这样，消费者可以按照自己的节奏进行消费，不受其他消费者的影响。</span></li>
   <li fid="ub938c195" data-lake-id="u079a2f47" id="u079a2f47"><span data-lake-id="u9419e52b" id="u9419e52b">数据的可靠性：高水位还用于确保数据的可靠性。在Kafka中，只有消息被写入主副本（Leader Replica）并被所有的同步副本（In-Sync Replicas，ISR）确认后，才被认为是已提交的消息。高水位表示已经被提交的消息的边界。只有高水位之前的消息才能被认为是已经被确认的，其他的消息可能会因为副本故障或其他原因而丢失。</span></li>
  </ul>
  <p data-lake-id="u7ecc8e0d" id="u7ecc8e0d"><span data-lake-id="u904ffb80" id="u904ffb80">​</span><br></p>
  <p data-lake-id="u67f18e09" id="u67f18e09"><span data-lake-id="u06a2ddaf" id="u06a2ddaf">还有一个概念，叫做LEO，即 Log End Offset，他是日志最后消息的偏移量。 它标识当前日志文件中下一条待写入消息的 offset。</span></p>
  <p data-lake-id="u1db6f313" id="u1db6f313"><span data-lake-id="ufd4240ad" id="ufd4240ad">​</span><br></p>
  <p data-lake-id="ua2830343" id="ua2830343"><img src="https://cdn.nlark.com/yuque/0/2023/png/5378072/1685778254153-71a9824b-b249-44b9-8368-e8b238a5dd16.png?x-oss-process=image%2Fwatermark%2Ctype_d3F5LW1pY3JvaGVp%2Csize_42%2Ctext_SmF2YSA4IEd1IFA%3D%2Ccolor_FFFFFF%2Cshadow_50%2Ct_80%2Cg_se%2Cx_10%2Cy_10"></p>
  <p data-lake-id="u1970690f" id="u1970690f"><br></p>
  <p data-lake-id="u2a75dd92" id="u2a75dd92"><span data-lake-id="u4043188a" id="u4043188a">当消费者消费消息时，它可以使用高水位作为参考点，只消费高水位之前的消息，以确保消费的是已经被确认的消息，从而保证数据的可靠性。如上图，只消费offet为6之前的消息。</span></p>
  <p data-lake-id="ue3e3bfef" id="ue3e3bfef"><span data-lake-id="u098ffa7e" id="u098ffa7e">​</span><br></p>
  <p data-lake-id="u0b9ebeec" id="u0b9ebeec"><span data-lake-id="uf8097637" id="uf8097637">​</span><br></p>
  <p data-lake-id="uda01adcf" id="uda01adcf"><span data-lake-id="uefd3df5b" id="uefd3df5b">我们都知道，在Kafka中，每个分区都有一个Leader副本和多个Follower副本。</span></p>
  <p data-lake-id="ucb8f3991" id="ucb8f3991"><span data-lake-id="u004917e9" id="u004917e9">​</span><br></p>
  <p data-lake-id="ua79b8084" id="ua79b8084"><span data-lake-id="u8e2b66aa" id="u8e2b66aa">当Leader副本发生故障时，Kafka会选择一个新的Leader副本。这个切换过程中，需要保证数据的一致性，即新的Leader副本必须具有和旧Leader副本一样的消息顺序。</span></p>
  <p data-lake-id="uf9449010" id="uf9449010"><span data-lake-id="u6c091e4e" id="u6c091e4e">​</span><br></p>
  <p data-lake-id="u10252199" id="u10252199"><span data-lake-id="ue075a3ba" id="ue075a3ba">为了实现这个目标，Kafka引入了Leader Epoch的概念。Leader Epoch是一个递增的整数，每次副本切换时都会增加。它用于标识每个Leader副本的任期。</span></p>
  <p data-lake-id="uc830069a" id="uc830069a"><span data-lake-id="u15f08a4e" id="u15f08a4e">​</span><br></p>
  <p data-lake-id="u67a078d9" id="u67a078d9"><span data-lake-id="u39a992e5" id="u39a992e5">每个副本都会维护自己的Leader Epoch记录。它记录了副本所属的分区在不同Leader副本之间切换时的任期。</span></p>
  <p data-lake-id="udab34691" id="udab34691"><span data-lake-id="u72384b7a" id="u72384b7a">​</span><br></p>
  <p data-lake-id="uf937e3be" id="uf937e3be"><span data-lake-id="ua018dc53" id="ua018dc53">在副本切换过程中，新的Leader会检查旧Leader副本的Leader Epoch和高水位。只有当旧Leader副本的Leader Epoch小于等于新Leader副本的Leader Epoch，并且旧Leader副本的高水位小于等于新Leader副本的高水位时，新Leader副本才会接受旧Leader副本的数据。</span></p>
  <p data-lake-id="ua1e2986f" id="ua1e2986f"><span data-lake-id="u5d0ac5cd" id="u5d0ac5cd">​</span><br></p>
  <p data-lake-id="ue2bb1114" id="ue2bb1114"><span data-lake-id="u2e6fec36" id="u2e6fec36">通过使用Leader Epoch和高水位的验证，Kafka可以避免新的Leader副本接受旧Leader副本之后的消息，从而避免数据回滚。只有那些在旧Leader副本的Leader Epoch和高水位之前的消息才会被新Leader副本接受。</span></p>
  <p data-lake-id="ucd63866d" id="ucd63866d"><span data-lake-id="u766d2e30" id="u766d2e30">​</span><br></p>
  <h1 data-lake-id="G8WNl" id="G8WNl"><span data-lake-id="u1e45e2c8" id="u1e45e2c8">扩展知识</span></h1>
  <p data-lake-id="ufcd0c40c" id="ufcd0c40c"><br></p>
  <h2 data-lake-id="vTNDl" id="vTNDl"><span data-lake-id="u8e098c26" id="u8e098c26">Leader Epoch的过程</span></h2>
  <p data-lake-id="ud67a9600" id="ud67a9600"><br></p>
  <p data-lake-id="u5df446c6" id="u5df446c6"><span data-lake-id="u1b1192b2" id="u1b1192b2">每个分区都有一个初始的Leader Epoch，通常为0。</span></p>
  <p data-lake-id="uc0b9fc7a" id="uc0b9fc7a"><span data-lake-id="u83e75e6d" id="u83e75e6d">​</span><br></p>
  <p data-lake-id="ubae14cf1" id="ubae14cf1"><span data-lake-id="u928a2e2d" id="u928a2e2d">当Leader副本发生故障或需要进行切换时，Kafka会触发副本切换过程。</span></p>
  <p data-lake-id="u6d09465f" id="u6d09465f"><span data-lake-id="ud9b26af5" id="ud9b26af5">​</span><br></p>
  <p data-lake-id="u033da2d1" id="u033da2d1"><span data-lake-id="u36e3c4f3" id="u36e3c4f3">副本切换过程中，Kafka会从ISR（In-Sync Replicas，同步副本）中选择一个新的Follower副本作为新的Leader副本。</span></p>
  <p data-lake-id="u4d95b8f0" id="u4d95b8f0"><span data-lake-id="u0ffa36d1" id="u0ffa36d1">​</span><br></p>
  <p data-lake-id="uffee6607" id="uffee6607"><span data-lake-id="u27b50497" id="u27b50497">新的Leader副本会增加自己的Leader Epoch，使其大于之前的Leader Epoch。这表示进入了一个新的任期。</span></p>
  <p data-lake-id="uacdd7ec4" id="uacdd7ec4"><span data-lake-id="uee309fc8" id="uee309fc8">​</span><br></p>
  <p data-lake-id="u330e9dce" id="u330e9dce"><span data-lake-id="u9c1e571d" id="u9c1e571d">新的Leader副本会验证旧Leader副本的状态以确保数据的一致性。它会检查旧Leader副本的Leader Epoch和高水位。</span></p>
  <p data-lake-id="u0f05a688" id="u0f05a688"><span data-lake-id="u9cef7d16" id="u9cef7d16">​</span><br></p>
  <p data-lake-id="u8b10accf" id="u8b10accf"><span data-lake-id="u6da9e2b3" id="u6da9e2b3">如果旧Leader副本的Leader Epoch小于等于新Leader副本的Leader Epoch，并且旧Leader副本的高水位小于等于新Leader副本的高水位，则验证通过。</span></p>
  <p data-lake-id="u92386954" id="u92386954"><span data-lake-id="ua38407fe" id="ua38407fe">​</span><br></p>
  <p data-lake-id="u17b02b56" id="u17b02b56"><span data-lake-id="u3f2fa141" id="u3f2fa141">一旦验证通过，新的Leader副本会开始从ISR中的一部分副本复制数据，以确保新Leader上的数据与旧Leader一致。</span></p>
  <p data-lake-id="ue97788a4" id="ue97788a4"><span data-lake-id="u4d2fb0ca" id="u4d2fb0ca">​</span><br></p>
  <p data-lake-id="u1cc91634" id="u1cc91634"><span data-lake-id="uee9613a3" id="uee9613a3">一旦新的Leader副本复制了旧Leader副本的所有数据，并达到了与旧Leader副本相同的高水位，副本切换过程就完成了。</span></p>
 </body>
</html>